package com.alison.source

import org.apache.spark.sql.types.{LongType, StringType, StructType}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

/**
 * link: https://www.hangge.com/blog/cache/detail_3362.html#google_vignette
 */
object E4_sql {

  def main(args: Array[String]): Unit = {
//    sqler
//    dsler
    dsl_dataset
  }

  def sqler() = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("sqler")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换

    // 定义 Schema, 用于指定列名以及列中的数据类型
    val peopleSchema: StructType = new StructType()
      .add("name", StringType)
      .add("age", LongType)
      .add("sex", StringType)

    // 创建一个流式DataFrame，这里从文件系统中读取数据
    val directory = this.getClass.getResource("/").toString
    val peopleDF: DataFrame = spark.readStream
      .format("json")
      .schema(peopleSchema)
      .load(directory + "/datas/person") // 必须是目录

    // 创建临时表
    peopleDF.createOrReplaceTempView("people")

    // 查询年龄大于20的所有人员信息
    val df: DataFrame = spark.sql("select * from people where age > 20")

    // 启动查询, 把结果打印到控制台
    val query = df.writeStream
      .outputMode("append") // 使用append输出模式
      .format("console")
      .start()

    // 等待应用程序终止
    query.awaitTermination()

    //关闭 Spark
    spark.stop()
  }
  def dsler() = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("dsler")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换

    // 定义 Schema, 用于指定列名以及列中的数据类型
    val peopleSchema: StructType = new StructType()
      .add("name", StringType)
      .add("age", LongType)
      .add("sex", StringType)

    // 创建一个流式DataFrame，这里从文件系统中读取数据
    val directory = this.getClass.getResource("/").toString
    val peopleDF: DataFrame = spark.readStream
      .format("json")
      .schema(peopleSchema)
      .load(directory + "/datas/person") // 必须是目录

    // 查询年龄大于20的所有人员信息
    val df: DataFrame = peopleDF.select("name", "age", "sex").where("age > 20")

    // 启动查询, 把结果打印到控制台
    val query = df.writeStream
      .outputMode("append") // 使用append输出模式
      .format("console")
      .start()

    // 等待应用程序终止
    query.awaitTermination()

    //关闭 Spark
    spark.stop()
  }

  // 定义一个People样例类
  case class People(name: String, age: Long, sex: String)

  def dsl_dataset() = {
    // 创建 SparkSession
    val spark = SparkSession.builder()
      .appName("Hello")
      .master("local[*]")
      .getOrCreate()

    // 导入隐式转换
    import spark.implicits._

    // 定义 Schema, 用于指定列名以及列中的数据类型
    val peopleSchema: StructType = new StructType()
      .add("name", StringType)
      .add("age", LongType)
      .add("sex", StringType)

    // 创建一个流式DataFrame，这里从文件系统中读取数据
    val directory = this.getClass.getResource("/").toString
    val peopleDF: DataFrame = spark.readStream
      .format("json")
      .schema(peopleSchema)
      .load(directory + "/datas/person") // 必须是目录

    // 转成 ds
    val peopleDS: Dataset[People] = peopleDF.as[People]

    // 查询年龄大于20的所有人员信息
    val ds: Dataset[String] = peopleDS.filter(_.age > 20).map(_.name)

    // 启动查询, 把结果打印到控制台
    val query = ds.writeStream
      .outputMode("append") // 使用append输出模式
      .format("console")
      .start()
    // 等待应用程序终止
    query.awaitTermination()
    //关闭 Spark
    spark.stop()
  }
}
